package cn.edu.flink.scala.tutorial.多流转换

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.OutputTag
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object SideOutputTest {
  val maryTag: OutputTag[Event] = OutputTag[Event]("Mary")
  val aliceTag: OutputTag[Event] = OutputTag[Event]("Alice")

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource)

    val processedStream = clickStream.process(new ProcessFunction[Event, Event] {
      override def processElement(value: Event, ctx: ProcessFunction[Event, Event]#Context, out: Collector[Event]): Unit = {
        if (value.user == "Mary") {
          ctx.output(maryTag, value)
        } else if (value.user == "Alice") {
          ctx.output(aliceTag, value)
        } else {
          out.collect(value)
        }
      }
    })

    processedStream.getSideOutput(maryTag).print("Mary")
    processedStream.getSideOutput(aliceTag).print("Alice")
    processedStream.print("other")

    env.execute("SideOutputTest")
  }
}
